package rpc.core;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.util.CollectionUtils;
import rpc.bootstrap.springboot.Bootstrap;
import rpc.core.annotations.Flow;
import rpc.core.plugin.impl.proxy.AsyncFlowService;
import rpc.log.LogFormatManager;
import rpc.trace.Trace;
import rpc.trace.TraceContext;
import rpc.utils.ExceptionUtils;
import rpc.utils.TypeSafe;

public abstract class BaseAsyncFlow {

  private final List<InvokeInfo> invokeInfoList = new ArrayList<>();
  protected Logger logger = LoggerFactory.getLogger(getClass());
  protected FlowReq req;
  private String currFlowName;
  private ApplicationContext context;
  private Environment environment;
  private ExecutorService executorService;
  private AsyncFlowService asyncFlowService;
  private CompletableFuture<FlowRes>[] callFutures;
  private CompletableFuture<FlowRes> resFuture;
  private Bootstrap bootstrap;
  private List<FlowRes> callFlowResList;
  private boolean autoReply = false;

  public BaseAsyncFlow setAutoReply(boolean autoReply) {
    this.autoReply = autoReply;
    return this;
  }

  public BaseAsyncFlow setAsyncFlowService(AsyncFlowService asyncFlowService) {
    this.asyncFlowService = asyncFlowService;
    return this;
  }

  public BaseAsyncFlow setReq(FlowReq req) {
    this.req = req;
    return this;
  }

  public BaseAsyncFlow setExecutorService(ExecutorService executorService) {
    this.executorService = executorService;
    return this;
  }

  public BaseAsyncFlow setContext(ApplicationContext context) {
    this.context = context;
    return this;
  }

  protected int intConfig(String key) {
    return intConfig(key, 0);
  }

  protected int intConfig(String key, int dv) {
    return TypeSafe.anyToInt(environment.getProperty(key, String.valueOf(dv)));
  }

  protected String strConfig(String key) {
    return strConfig(key, null);
  }

  protected String strConfig(String key, String dv) {
    return TypeSafe.anyToString(environment.getProperty(key, dv));
  }

  public BaseAsyncFlow setBootstrap(Bootstrap bootstrap) {
    this.bootstrap = bootstrap;
    return this;
  }

  protected InvokeInfo invokeFeature(String flowName, FlowReq req) {
    return new InvokeInfo(flowName, req);
  }

  protected void invoke(Runnable callBack, String flowName, FlowReq req) {
    this.invoke(callBack, new InvokeInfo(flowName, req));
  }

  protected void invoke(Runnable callBack, InvokeInfo... invokeInfos) {
    this.invoke(callBack, Arrays.asList(invokeInfos));
  }

  @SuppressWarnings("unchecked")
  protected void invoke(Runnable callBack, List<InvokeInfo> invokeInfos) {
    int invokeSize = invokeInfos.size();
    TraceContext traceContext = Trace.get();
    CompletableFuture<FlowRes>[] futureList = new CompletableFuture[invokeSize];
    for (int i = 0; i < invokeSize; i++) {
      InvokeInfo invokeInfo = invokeInfos.get(i);
      String flowName = invokeInfo.getFlowName().toLowerCase();
      futureList[i] = bootstrap.invokeFlow(flowName, invokeInfo.getReq());
    }
    this.callFutures = futureList;
    try {

      CompletableFuture.allOf(callFutures).thenRun(() -> {
        Trace.resetTraceContext(traceContext);
        List<FlowRes> resList = new ArrayList<>();
        callFlowResList = new ArrayList<>(callFutures.length);
        for (int i = 0; i < invokeSize; i++) {
          CompletableFuture<FlowRes> f = callFutures[i];
          try {
            resList.add(f.get(req.getTimeOut(), TimeUnit.MILLISECONDS));
          } catch (Exception e) {
            logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
            resList.add(new FlowRes().fail(-827).msg("e:" + e.getMessage()));
          }
        }
        this.callFlowResList = resList;
        List<FlowRes> resFailList = this.callFlowResList.stream().filter(FlowRes::isFail).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(invokeInfos) && autoReply) {
          reply(resFailList.get(0));
          return;
        }
        callBack.run();
      });
    } catch (Exception e) {
      logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
    }
  }

  protected void reply(FlowRes res) {
    if (!this.resFuture.isDone()) {
      resFuture.complete(res);
      LogFormatManager.logger(currFlowName, req, res);
    } else {
      logger.error("reply res: repeat ,throw:" + res.isSucc() + "-" + res.retMsg() + "-" + res.retMsg());
    }
  }

  public BaseAsyncFlow setEnvironment(Environment environment) {
    this.environment = environment;
    return this;
  }

  protected void fail(int retCode) {
    fail(retCode, "");
  }

  protected void fail(int retCode, String msg) {
    reply(new FlowRes().fail(retCode).msg(msg));
  }

  protected void delay(String flowName, FlowReq req) {
    addDelayedTask(flowName, req);
  }

  protected void post(String flowName, FlowReq req) {
    this.bootstrap.invokeFlow(flowName, req).thenAccept(res -> {
      addDelayedTask(flowName, req);
      LogFormatManager.logger(currFlowName, req, res);
    });
  }

  public void addDelayedTask(String flowName, FlowReq req) {
    this.asyncFlowService.addDelayedTask(flowName, req);
  }

  //  private List<FlowRes> callFlowResList;
  protected void reply() {
    reply(new FlowRes());
  }

  protected void reply(int retCode) {
    reply(new FlowRes().fail(retCode));
  }

  protected void reply(int retCode, String msg) {
    reply(new FlowRes().fail(retCode).msg(msg));
  }

  public CompletableFuture<FlowRes> handle(FlowReq req) {
    TraceContext traceContext = Trace.get();
    CompletableFuture<FlowRes> future = new CompletableFuture<>();
    this.resFuture = future;
    this.req = req;

    this.executorService.execute(() -> {
      try {
        Trace.resetTraceContext(traceContext);
        this.enter();
      } catch (Exception e) {
        logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
        logger.info("exec error:{}", e.getMessage());
        fail(-8279, "e:" + e.getMessage());
      }
    });
    return future;
  }

  public abstract void enter();

  public void retRes() {
    FlowRes lastResult = getLastResult();
    reply(lastResult);
  }

  protected FlowRes getLastResult() {
    return getLastResults().get(0);
  }

  protected List<FlowRes> getLastResults() {
    return this.callFlowResList;
  }

  public void init() {
    this.currFlowName = this.getClass().getAnnotation(Flow.class).value();
  }
}
